package com.yifeng.repo.flink.data.transport.streaming.functions.task;


import java.sql.PreparedStatement;
import java.util.HashMap;
import java.util.List;

import com.yifeng.repo.flink.data.transport.config.DataSourceConfig;
import com.yifeng.repo.flink.data.transport.dto.DataPacket;
import com.yifeng.repo.flink.data.transport.dto.DataRow;

/**
 * mysql全量同步的任务
 * @author lijing
 * @param <T>
 * @since 2023-01-03
 *
 */
public class MysqlFullSyncTask<T> extends CommonFullSyncTask {

	public MysqlFullSyncTask(DataSourceConfig dataSourceConfig) {
		super(dataSourceConfig);
	}

	@Override
	protected PreparedStatement setFetchDirection(PreparedStatement ps, int fetchSize) throws Exception {
		ps.setFetchSize(Integer.MIN_VALUE);
		return ps;
	}

	@Override
	public DataRow assembleDataRow(DataRow.Type type,String tableName, HashMap dataMap) {
		return new DataRow(type, dataMap, "",tableName,dataSourceConfig.getDatabase());
	}

	@Override
	public DataPacket assembleDataPacket(int queueNo, long idx, List rows, String tableName, boolean end) {
		return DataPacket.ofFull(queueNo,idx, rows,tableName,dataSourceConfig.getDatabase(),"",dataSourceConfig.getHostname(),end);
	}

}
